package com.jiayuan.cn.energy.test;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.AggregateOperator;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.util.Collector;

import java.util.Arrays;
import java.util.Collection;

/*
 * Flink remote任务运行
 *
 * @author zhangwei
 * @date 2021/10/29 14:00
 * @since 1.0.0
 * @version 1.0.0
 */
public class HelloFlink {
    public static final class Tokenizer implements FlatMapFunction<String, Tuple2<String, Integer>> {
        @Override
        public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
//            Arrays.stream(value.toLowerCase().split("\\W+")).filter(word -> !word.isEmpty()).forEach(word -> out.collect(new Tuple2<>(word, 1)));
            Arrays.stream(value.toLowerCase().split("\\s")).filter(word -> !word.isEmpty()).forEach(word -> out.collect(new Tuple2<>(word, 1)));
        }
    }

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> text = env.fromData("Hello flink", "Hello world", "Hello love");
        DataStream<Tuple2<String, Integer>> counts = text.flatMap(new Tokenizer()).keyBy(0).sum(1);
        System.out.println("----------------------------------------------------");
        counts.print();
        env.execute("Wordcount");
    }
}
